package com.atliuzu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atliuzu.bean.CartAddUuBean;
import com.atliuzu.bean.TrafficBehaviorOfTheFunnelBean;
import com.atliuzu.utils.ClickHouseUtil;
import com.atliuzu.utils.DateFormatUtil;
import com.atliuzu.utils.MyKafkaUtil;
import lombok.Value;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author w
 * @create 2022-08-21-21:11
 */
public class DwsUserPageHomeDetailWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment ( );
        env.setParallelism ( 1 );

        //TODO 2.状态后端设置
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                3, Time.days(1), Time.minutes(1)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage(
//                "hdfs://hadoop102:8020/ck"
//        );
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //TODO 3.读取DWD数据
        String PageTopic = "dwd_traffic_page_log";
        String groupId = "Behavior_Of_The_Funnel";
        DataStreamSource<String> pageSourceDS = env.addSource ( MyKafkaUtil.getFlinkKafkaConsumer ( PageTopic, groupId ) );

        //TODO 4.将数据转换为JSON对象,并过率数据,因需要过滤filter和转换map
        SingleOutputStreamOperator<JSONObject> pageDS = pageSourceDS.flatMap ( new FlatMapFunction<String, JSONObject> ( ) {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject ( value );
                String pageId = jsonObject.getJSONObject ( "page" ).getString ( "page_id" );
                if ("home".equals ( pageId ) || "good_detail".equals ( pageId )) {
                    out.collect ( jsonObject );
                }
            }
        } );

        //TODO 5.按照uid分组
        KeyedStream<JSONObject, String> pageKeyedStream = pageDS.keyBy ( json -> json.getJSONObject ( "common" ).getString ( "uid" ) );

        //TODO 6.去重,将数据转换为JavaBean(page,cart,order,pay)
        SingleOutputStreamOperator<TrafficBehaviorOfTheFunnelBean> pageUvDS = pageKeyedStream.flatMap ( new RichFlatMapFunction<JSONObject, TrafficBehaviorOfTheFunnelBean> ( ) {

            //定义状态
            private ValueState<String> homeState;
            private ValueState<String> detailState;

            @Override
            public void open(Configuration parameters) throws Exception {
                //设置TTL时间和更新方式为写时重置
                StateTtlConfig ttlConfig = new StateTtlConfig.Builder ( Time.days ( 1 ) ).setUpdateType ( StateTtlConfig.UpdateType.OnCreateAndWrite ).build ( );

                ValueStateDescriptor<String> homeDesc = new ValueStateDescriptor<> ( "home-state", String.class );
                ValueStateDescriptor<String> detailDesc = new ValueStateDescriptor<> ( "detail-state", String.class );

                //设置TTL
                homeDesc.enableTimeToLive ( ttlConfig );
                detailDesc.enableTimeToLive ( ttlConfig );
                //初始化状态
                homeState = getRuntimeContext ( ).getState ( homeDesc );
                detailState = getRuntimeContext ( ).getState ( detailDesc );
            }

            @Override
            public void flatMap(JSONObject value, Collector<TrafficBehaviorOfTheFunnelBean> out) throws Exception {
                //获取状态日期以及当前数据的日期以及页面ID
                String homeLastDT = homeState.value ( );
                String detailLastDT = detailState.value ( );
                String currentDT = DateFormatUtil.toDate ( value.getLong ( "ts" ) );
                String pageID = value.getJSONObject ( "page" ).getString ( "page_id" );

                //设置初始状态值为0
                long homeUvCt = 0L;
                long goodDetailUvCt = 0L;

                if ("home".equals ( pageID ) && (homeLastDT == null || !homeLastDT.equals ( currentDT ))) {
                    homeUvCt = 1L;
                    homeState.update ( currentDT );
                } else if ("good_detail".equals ( pageID ) && (detailLastDT == null || !detailLastDT.equals ( currentDT ))) {
                    goodDetailUvCt = 1L;
                    detailState.update ( currentDT );
                }
                //输出数据,判断过滤为0的数据
                if (homeUvCt == 1L || goodDetailUvCt ==1L ) {
                    out.collect ( new TrafficBehaviorOfTheFunnelBean (
                            "",
                            "",
                            homeUvCt,
                            goodDetailUvCt,
                            value.getLong ( "ts" )
                    ) );
                }
            }
        } );


        //TODO 设置watermark
        SingleOutputStreamOperator<TrafficBehaviorOfTheFunnelBean> watermarkDS = pageUvDS.assignTimestampsAndWatermarks ( WatermarkStrategy.<TrafficBehaviorOfTheFunnelBean>forBoundedOutOfOrderness ( Duration.ofSeconds ( 2 ) ).withTimestampAssigner ( new SerializableTimestampAssigner<TrafficBehaviorOfTheFunnelBean> ( ) {
            @Override
            public long extractTimestamp(TrafficBehaviorOfTheFunnelBean element, long recordTimestamp) {
                return element.getTs ( );
            }
        } ) );

        //todo 开窗聚合
        SingleOutputStreamOperator<TrafficBehaviorOfTheFunnelBean> reduceDS = watermarkDS.windowAll ( TumblingEventTimeWindows.of ( org.apache.flink.streaming.api.windowing.time.Time.seconds ( 10L ) ) ).reduce ( new ReduceFunction<TrafficBehaviorOfTheFunnelBean> ( ) {
            @Override
            public TrafficBehaviorOfTheFunnelBean reduce(TrafficBehaviorOfTheFunnelBean value1, TrafficBehaviorOfTheFunnelBean value2) throws Exception {
                value1.setHomeUvCt ( value1.getHomeUvCt () + value2.getHomeUvCt () );
                value1.setGoodDetailUvCt ( value1.getGoodDetailUvCt () + value2.getGoodDetailUvCt () );
                return value1;
            }
        }, new AllWindowFunction<TrafficBehaviorOfTheFunnelBean, TrafficBehaviorOfTheFunnelBean, TimeWindow> ( ) {
            @Override
            public void apply(TimeWindow window, Iterable<TrafficBehaviorOfTheFunnelBean> values, Collector<TrafficBehaviorOfTheFunnelBean> out) throws Exception {
                TrafficBehaviorOfTheFunnelBean next = values.iterator ( ).next ( );
                //补充信息
                next.setTs ( System.currentTimeMillis ( ) );
                next.setEdt ( DateFormatUtil.toYmdHms ( window.getEnd ( ) ) );
                next.setStt ( DateFormatUtil.toYmdHms ( window.getStart ( ) ) );

                //输出数据
                out.collect ( next );
            }
        } );

        reduceDS.print ( "reduce>>>>>" );

        //写入clickHouse
        SinkFunction<TrafficBehaviorOfTheFunnelBean> jdbcSink = ClickHouseUtil.<TrafficBehaviorOfTheFunnelBean>getJdbcSink(
                "insert into dws_traffic_page_view_window values(?,?,?,?,?)"
        );


        //TODO 9.启动任务
        env.execute ( "DwsTrafficPageViewWindow" );
    }
}
